package com.demo.component.srv;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

/**
 * Author:  Hardy
 * Date:    2018/7/18 20:05
 * Description: RabbitMQ 发送组件 同步发送且同步确认
 **/
@Component
public class RabbitSender {

    private static final Logger log = LoggerFactory.getLogger(RabbitSender.class);

    private final int retryCount;
    private final long retryInterval;
    private final RabbitTemplate rabbitTemplate;
    private final RabbitTransactionManager rabbitTransactionManager;

    @Autowired
    public RabbitSender(@Value("${rabbit.retry.count:3}") int count,
                        @Value("${rabbit.retry.interval:200}") long interval,
                        ConnectionFactory connectionFactory,
                        MessageConverter messageConverter) {
        /**
         * @Value("${rabbit.retry.count:3}") 获取配置rabbit.retry.count,获取不到时，取默认值为3, 没有':3', 获取不到时,项目启动异常
         */
        this.retryCount = count;
        this.retryInterval = interval;
        /**
         *  不初始化 @Bean 原因：
         *      (1) RabbitTransactionManager 和 DbTransaction 都实现 PlatformTransactionManager @Bean 初始化 RabbitTransactionManager 可能导致 @Transactional db 不可用
         *      (2) 如果引用 sping-data-jpa @Transactional 导致 @Transactional 同时开启管理 Rabbit、Db 事务，增加了复杂度
         *      (3) @Transactional 同时开启 Rabbit、Db 事务，但是实质是2个事务，逐一事务提交，假如事务1提交成功，事务2提交失败，可能会造成难以排查的问题
         */
        this.rabbitTransactionManager = new RabbitTransactionManager(connectionFactory);
        /**
         * 不直接注入 RabbitTemplate：
         *      (1) 增加兼容性不影响 spring 原有单例 RabbitTemplate
         */
        this.rabbitTemplate = new RabbitTemplate(connectionFactory);
        this.rabbitTemplate.setMessageConverter(messageConverter);
        this.rabbitTemplate.setChannelTransacted(true);
    }

    public void sendAndRetry(String exchange, String routingKey, Object object) {
        int retry = 0;
        while (true) {
            try {
                send(exchange, routingKey, object);
                return;
            } catch (Exception e) {
                if (++retry > retryCount)
                    throw e;
                sleep();
                log.info("Retry {} time publish message {}", retry, object.toString());
            }
        }
    }

    private void send(String exchange, String routingKey, Object object) {
        DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
        definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
        definition.setName("RabbitTool.sendPvt");

        TransactionStatus status = rabbitTransactionManager.getTransaction(definition);
        try {
            rabbitTemplate.convertAndSend(exchange, routingKey, object);
            rabbitTransactionManager.commit(status);
        } catch (Exception e) {
            rabbitTransactionManager.rollback(status);
            throw e;
        }
    }

    private void sleep() {
        try {
            Thread.sleep(retryInterval);
        } catch (InterruptedException e) {
            log.warn("Retry interval Interrupted");
        }
    }
}